package com.lagou.hbase.client;

import com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class HbaseClientDemo {
    Configuration conf = null;
    Connection conn = null;

    private Map<String, List<String>> relationshipMap = new HashMap<>();

    {
        relationshipMap.put("uid1", Lists.newArrayList("uid2","uid5","uid7"));
        relationshipMap.put("uid2", Lists.newArrayList("uid1","uid3","uid6","uid10"));
        relationshipMap.put("uid3", Lists.newArrayList("uid2","uid4","uid7"));
    }

    @Before
    public void init() throws IOException {
        //获取一个配置文件对象
        conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        //通过conf获取到hbase集群的连接
        conn = ConnectionFactory.createConnection(conf);
    }

    //创建一张hbase表
    @Test
    public void createTable() throws IOException {
        //获取HbaseAdmin对象用来创建表
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        //创建Htabledesc描述器，表描述器
        final HTableDescriptor relationship = new HTableDescriptor(TableName.valueOf("relationship"));
        //指定列族
        relationship.addFamily(new HColumnDescriptor("friends"));
        admin.createTable(relationship);
        System.out.println("relationship表创建成功！！");
    }

    private void breakup(String uid1, String uid2) throws IOException {
        final Table relationship = conn.getTable(TableName.valueOf("relationship"));
        final Delete delete1 = new Delete(Bytes.toBytes(uid1));
        delete1.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(uid2));
        relationship.delete(delete1);

        final Delete delete2 = new Delete(Bytes.toBytes(uid2));
        delete2.addColumn(Bytes.toBytes("friends"),Bytes.toBytes(uid1));
        relationship.delete(delete2);
        //关闭table对象
        relationship.close();
    }

    @Test
    public void putData() throws IOException {
        //需要获取一个table对象
        final Table relationship = conn.getTable(TableName.valueOf("relationship"));

        relationshipMap.entrySet().forEach(entry->{
            String rowKey = entry.getKey();
            entry.getValue().forEach(friend->{
                Put put = new Put(Bytes.toBytes(rowKey));
                put.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(friend), Bytes.toBytes(friend));
                try {
                    relationship.put(put);
                    relationship.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
            System.out.println("插入数据成功！！"+entry);
        });
    }

    @Test
    public void deleteData() throws IOException {
        breakup("uid1","uid2");
    }

    //释放连接
    @After
    public void realse() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
